package com.yequan.mq.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author yq
 * @date 2020/09/01
 */
public class ProducerBroadCast {

    public static void main(String[] args) {
        DefaultMQProducer producer = new DefaultMQProducer("push_consumer");
        producer.setNamesrvAddr("127.0.0.1:9876");
        try {
            // 设置实例名称
//            producer.setInstanceName("producer_broadcast");
            // 设置重试次数
//            producer.setRetryTimesWhenSendFailed(3);
            // 开启生产者
            producer.start();
            AtomicInteger i = new AtomicInteger(0);
            for (; ; ) {
                // 创建一条消息
                Message msg = new Message("topic_broadcast", "TagA", "OrderID0034", ("message_broadcast_test--->" + (i.getAndIncrement())).getBytes());
                SendResult send = producer.send(msg);
                System.out.println("id:--->" + send.getMsgId() + ",result:--->" + send.getSendStatus());
                Thread.sleep(2000);
                if (i.get() == 10) {
                    return;
                }
            }
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        producer.shutdown();
    }

}
